package ds_industry_2025.ds.ds_07.T4
/*
    2、根据第1小题的结果，对其进行聚合，其中对sku_id进行one-hot转换，将其转换为以下格式矩阵：第一列为用户id，其余列名为商品id，
    按照用户id进行升序排序，展示矩阵第一行前5列数据，将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。
字段	类型	中文含义	备注
user_id	double	客户key
sku_id0	double	用户是否购买过商品1	若用户购买过该商品，则值为1，否则为0
sku_id1	double	用户是否购买过商品2	若用户购买过该商品，则值为1，否则为0
sku_id2	double	用户是否购买过商品3	若用户购买过该商品，则值为1，否则为0
.....
结果格式如下：
---------------第一行前5列结果展示为---------------
0.0,1.0,0.0,0.0,0.0
 */

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, SparkSession}

import java.util.Properties
object t2 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t1")
      .config("hive.exec.dynamic.partition.mode", "nonstrict")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    val conn = new Properties()
    conn.setProperty("user", "root")
    conn.setProperty("password", "123456")
    conn.setProperty("driver", "com.mysql.jdbc.Driver")

    //  todo 创建读取mysql数据的方法
    def read(tablename: String): DataFrame = {
      spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", tablename, conn)
    }

    //  todo 读取mysql数据
    val sku_info = read("sku_info")
    val user_info = read("user_info")
    val order_info = read("order_info")
    val order_detail = read("order_detail")

    //  todo 剔除订单表和详细订单表里面user_id和sku_id不存在当前维度表里面的数据
    val skus = sku_info.select("id").distinct().withColumnRenamed("id", "id_1")
    val users = user_info.select("id").distinct().withColumnRenamed("id", "id_1")
    val order = order_info.join(users, users("id_1") === order_info("user_id"), "inner")
    val detail = order_detail.join(skus, skus("id_1") === order_detail("sku_id"), "inner")

    //  todo 对数据进行索引化
    val r1 = order.join(detail, detail("order_id") === order("id"))
      .select("user_id", "sku_id")
      .distinct()
      .withColumn("user_id", dense_rank().over(Window.orderBy("user_id")) - 1)
      .withColumn("sku_id", dense_rank().over(Window.orderBy("sku_id")) - 1)
      .orderBy("user_id", "sku_id")


    println("----------------user_id_mapping和sku_id_mapping数据前5条如下")
    r1.limit(5).collect().foreach(
      r => println(r(0) + ":" + r(1))
    )


    //  todo 接下来开始第二题
    //  将sku_id列的数据转化成sku_id+n
    val source=r1
      .withColumn("sku_id",concat(lit("sku_id"),col("sku_id").cast("String")))

    //  todo 提取所有的sku_id，为了后续pivot函数转换提供值使用
    val sku_lit:Array[String] = source.select("sku_id")
      .distinct()
      .orderBy(
        split(col("sku_id"), "id")(1).cast("int")
      )
      .map(_.toString())
      .collect()

    val result=source.groupBy("user_id")
      .pivot("sku_id",sku_lit)
      .agg(lit(1.0))
      .na.fill(0.0)
      .orderBy("user_id")
      .limit(1)

    println("---------------第一行前5列结果展示为----------------")
    result.select(result.columns.take(5).map(col): _*)
      .foreach(
        r => println(r.mkString(","))
      )


    spark.close()

  }

}
